package cn.com.tt.flink.sql.engine.pojo.sql;

import cn.com.bluemoon.bd.utils.RegexUtils;
import org.apache.flink.table.api.SqlDialect;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class BmUseDialectSql extends AbstractBmFlinkSql {
    private static final String HIVE_DIALECT = "hive";
    private static final String DEFAULT_DIALECT = "default";

    public BmUseDialectSql(String sqlStatement, StreamTableEnvironment tableEnv) {
        super(sqlStatement, tableEnv);
    }

    @Override
    protected void innerProcess() {
        String dialect = getDialect(sqlStatement);

        if (dialect.equals(HIVE_DIALECT)) {
            tableEnv.getConfig().setSqlDialect(SqlDialect.HIVE);
        } else if (dialect.equals(DEFAULT_DIALECT)) {
            tableEnv.getConfig().setSqlDialect(SqlDialect.DEFAULT);
        }
    }

    private String getDialect(String sqlStatement) {
        String trimSql = sqlStatement.trim().toLowerCase().replaceAll("\\s+", " ");
        return RegexUtils.extractData(trimSql, "=(.*?);").trim();
    }
}
